package com.xjrh.traintravel.collector.netty;

import com.alibaba.fastjson.JSONObject;
import com.xjrh.common.core.utils.FileUploadUtils;
import com.xjrh.middleware.redis.service.RedisDataDirectionEnum;
import com.xjrh.middleware.redis.service.RedisService;
import com.xjrh.traintravel.collector.api.BusinessTrainService;
import com.xjrh.traintravel.collector.common.CRC16M;
import com.xjrh.traintravel.collector.common.StringUtil;
import com.xjrh.traintravel.collector.mq.KafkaProducer;
import com.xjrh.traintravel.collector.mq.msg.*;
import com.xjrh.traintravel.collector.netty.coder.command.AgreementCommandEnum;
import com.xjrh.traintravel.collector.netty.coder.command.AgreementStructure;
import com.xjrh.traintravel.collector.netty.command.check.CheckApply;
import com.xjrh.traintravel.collector.netty.command.check.CheckData;
import com.xjrh.traintravel.collector.netty.command.data.DataContent;
import com.xjrh.traintravel.collector.netty.command.data.DataFileName;
import com.xjrh.traintravel.collector.netty.command.fault.EquipmentFault;
import com.xjrh.traintravel.collector.netty.command.fault.PartFault;
import com.xjrh.traintravel.collector.netty.command.front.DspData;
import com.xjrh.traintravel.collector.netty.command.front.FrontData;
import com.xjrh.traintravel.collector.netty.command.front.VersionData;
import com.xjrh.traintravel.collector.netty.command.mvb.MvbData;
import com.xjrh.traintravel.collector.netty.command.mvb.MvbStation;
import com.xjrh.traintravel.collector.netty.command.station.TravelStationRegionData;
import com.xjrh.traintravel.collector.netty.command.upgrade.UpgradeCommonApply;
import com.xjrh.traintravel.collector.netty.command.upgrade.UpgradeFileApply;
import com.xjrh.traintravel.collector.netty.command.upgrade.UpgradeProgress;
import com.xjrh.traintravel.collector.netty.handler.*;
import com.xjrh.traintravel.collector.runner.DataCacheTool;
import com.xjrh.traintravel.collector.runner.NettyClientManager;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 服务消息处理器
 *  <p>
 *      核心处理类。进行连接、消息的处理
 *  </p>
 * @author snps
 * @create 2022-01-17 11:10
 * @describe
 **/
@Component
@ChannelHandler.Sharable
@Slf4j
public class ServerHandler extends SimpleChannelInboundHandler<AgreementStructure> {

    /**
     * 数据缓存时间（单位：分钟）
     */
    @Value("${project.dataCacheTime}")
    private long dataCacheTime;

    /**
     * 上传采集问价目录
     */
    @Value("${project.uploadFilePath}")
    private String uploadFilePath;

    @Resource
    private AccessAgreementHandler accessAgreementHandler;

    @Resource
    private ConfigHandler configHandler;

    @Resource
    private ConfigReadHandler configReadHandler;

    @Resource
    private UpgradeHandler upgradeHandler;

    @Resource
    private DataHandler dataHandler;

    @Resource
    private FrontHandler frontHandler;

    @Resource
    private CheckHandler checkHandler;

    @Resource
    private MvbHandler mvbHandler;

    @Resource
    private FaultHandler faultHandler;

    @Resource
    private DataCacheTool dataCacheTool;

    @Resource
    private NettyClientManager nettyClientManager;

    @Resource
    private RedisService redisService;

    @Resource
    private BusinessTrainService businessTrainService;

    @Resource
    private KafkaProducer kafkaProducer;


    /**
     * 客户端与服务端创建连接的时候调用
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        log.warn("收到-客户端接入请求。【连接地址：{}】", channel.remoteAddress());
        channel.read();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, AgreementStructure agreement) throws Exception {
        if (agreement == null) {
            return;
        }

        // 进行CRC验证
        String calculateCrcValue = CRC16M.calculateCrc(agreement);
        if (!CRC16M.checkCrcCorrect(agreement.getFcs(), calculateCrcValue)) {
            AttributeKey<String> attrEquipmentId = AttributeKey.valueOf(BaseAgreementHandler.ATTR_KEY_EQUIPMENT_ID);
            ctx.channel().attr(attrEquipmentId).set(agreement.getAddr());
            log.error("客户端【{}】，校验错误！收到的FCS：{}，计算的FCS：{}", agreement.getAddr(), agreement.getFcs(), calculateCrcValue);
            return;
        }

        // 更新客户端连接时间
        nettyClientManager.saveClientChannelTime(agreement.getAddr(), new Date());

        // 根据消息进行业务处理
        doBusiness(agreement, ctx.channel());
    }

    /**
     * 服务端接收客户端发送过来的数据结束之后调用
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // ctx.flush();
    }

    /**
     * 客户端与服务端断开连接时调用
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();

        String equipmentId = accessAgreementHandler.getClientChannelKey(channel);
        log.warn("channelInactive, channelId: {}", equipmentId);
        if (StringUtil.isEmpty(equipmentId)) {
            log.warn("收到-客户端关闭请求。【连接地址：{}】", channel.remoteAddress());
        } else {
            log.warn("收到-客户端关闭请求。【连接地址：{}，设备标识：{}】", channel.remoteAddress(), equipmentId);

            // 修改客户端在线状态
            try {
                businessTrainService.setNetStatus(equipmentId, 0, null);
            } catch (Exception e) {
                log.error("修改客户端在线状态，失败！");
            }
            log.warn("设备【{}】离线", equipmentId);
        }

        // channel.close();
        closeChannel(ctx);
    }

    /**
     * 服务出现异常的时候调用
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        /*log.error("Netty Server Exception !!!");
        cause.printStackTrace();
        ctx.close();*/

        // 移除客户端连接
        String equipmentId = accessAgreementHandler.getClientChannelKey(ctx.channel());
        if (StringUtil.isEmpty(equipmentId)) {
            accessAgreementHandler.removeClientChannel(equipmentId);
        }

        closeChannel(ctx);
    }

    /**
     * 关闭通道并重新连接
     *
     * @param ctx
     */
    public void closeChannel(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();

        try {
            if (channel != null) {
                channel.close();
                // channel.eventLoop().shutdownGracefully();
            }
            ctx.close();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            channel = null;
        }
    }


    /***************************************************
     * 解析消息，执行业务处理
     * @param agreement 协议数据
     * @param channel
     */
    private void doBusiness(AgreementStructure agreement, Channel channel) {
        if (agreement == null) {
            log.error("协议数据为空，不进行业务处理!");
            return;
        }
        if (channel == null) {
            log.error("客户端连接通道为空，不进行业务处理!");
            return;
        }

        // 获取 控制位 内容
        String fc = agreement.getFc();
        if (StringUtil.isEmpty(fc)) {
            log.error("指令-控制位为空，不进行业务处理!");
            return;
        }

        /* 心跳 (00) */
        if (fc.equals(AgreementCommandEnum.S_00.getValue())) {
            // 收到：心跳
            accessAgreementHandler.heartbeat(agreement, channel);

            // 回复: 心跳响应
            accessAgreementHandler.replyHeartbeat(agreement, channel);
            return;
        }
        /* 登录 (01) */
        if (fc.equals(AgreementCommandEnum.S_01.getValue())) {
            if (agreement.getData().startsWith("01")) {
                // 收到：登录
                accessAgreementHandler.login(agreement, channel);

                // 回复: 当前时间
                accessAgreementHandler.replyLogin(agreement, channel);
                return;
            }
            return;
        }
        /* 配置参数#回复 (82) */
        if (fc.equals(AgreementCommandEnum.R_82.getValue())) {
            configHandler.getApplyCommand(agreement);
            return;
        }
        /* 读取配置参数#回复 (83) */
        if (fc.equals(AgreementCommandEnum.R_83.getValue())) {
            // 接收到的 字符串数据 转为 对应的配置命令
            configReadHandler.handleReplyData(agreement);
            return;
        }
        /* 采集数据上传 (04) */
        if (fc.equals(AgreementCommandEnum.S_04.getValue())) {
            // 收到：采集数据上传-文件名
            if (agreement.getData().startsWith("01")) {
                DataFileName dataFileName = dataHandler.getDataFileName(agreement.getAddr(), agreement.getData());

                // 在channel中保存文件名（写入文件内容时，需要从channel中获取文件名）
                AttributeKey<DataFileName> attrDataFileName = AttributeKey.valueOf("dataFileName");
                channel.attr(attrDataFileName).set(dataFileName);

                // 把文件名放入Redis中（使用Redis的过期机制，判断文件是否已经存在）
                String rootPath = uploadFilePath;
                // String businessPath = agreement.getAddr() + "/" + DateUtils.getDate2String() + "/" + dataFileName.getFileType() + "/";
                String businessPath = agreement.getAddr() + "/" + dataFileName.getCollectionDate() + "/" + dataFileName.getFileType() + "/";
                String originalFilePath = channel.attr(attrDataFileName).get().getDataFileName();
                originalFilePath = businessPath + originalFilePath;
                dataFileName.setOriginalFilePath(originalFilePath);
                // log.warn("保存数据文件: {}", originalFilePath);

                // 判断文件是否存在
                boolean fileExists = FileUploadUtils.isExists(rootPath + originalFilePath);
                if (fileExists) {
                    // TODO 2025-04-10 去掉：如果文件已存在，删除文件的操作
                    /*log.warn("文件已存在, 删除文件...");
                    FileUploadUtils.deleteFile(rootPath + originalFilePath);*/
                }
                // 创建文件
                try {
                    FileUploadUtils.createFile(rootPath + originalFilePath);
                } catch (IOException e) {
                    log.error("创建文件失败，文件路径：{}", rootPath + originalFilePath);
                    e.printStackTrace();
                }

                // 回复
                dataHandler.replyGetDataFileName(agreement, channel);
                return;
            }
            // 收到：采集数据上传-文件内容（数据）
            if (agreement.getData().startsWith("02")) {
                DataContent dataContent = dataHandler.getDataContent(agreement.getAddr(), agreement.getData());

                // 处理文件内容
                AttributeKey<DataFileName> attrDataFileName = AttributeKey.valueOf("dataFileName");
                DataFileName dataFileName = channel.attr(attrDataFileName).get();
                String fileName = channel.attr(attrDataFileName).get().getOriginalFilePath();
                boolean flag = dataHandler.handleDataContent(uploadFilePath + fileName, dataContent.getData());

                // 回复
                dataHandler.replyGetDataContent(agreement, flag?"00":"02", channel);

                // 如果接收文件结束（发送控制数据为：02），向Kafka发送消息
                if (dataContent.getLastFrameFlag().equals("02")) {
                    // 文件类型为：1-振动，3-温度，5-速度，向”队列-采集数据“发送消息
                    if (dataFileName.getFileType() == 1 || dataFileName.getFileType() == 3 || dataFileName.getFileType() == 5) {
                        CollectionDataMsg collectionDataMsg = new CollectionDataMsg();
                        collectionDataMsg.setEquipmentNo(dataFileName.getEquipmentId());
                        collectionDataMsg.setCom(dataFileName.getCom());
                        collectionDataMsg.setFrontType(dataFileName.getFrontType());
                        collectionDataMsg.setFrontAccess(dataFileName.getFrontAccess());
                        collectionDataMsg.setTemperature(dataFileName.getTemperature());
                        collectionDataMsg.setFileType(dataFileName.getFileType());
                        collectionDataMsg.setOriginalFile(fileName);
                        kafkaProducer.sendCollectionData(collectionDataMsg);
                    } else if (dataFileName.getFileType() == 2) {   // 文件类型为：2-波磨，向”队列-波磨“发送消息
                        WaveDataMsg waveDataMsg = new WaveDataMsg();
                        waveDataMsg.setEquipmentNo(dataFileName.getEquipmentId());
                        waveDataMsg.setCom(dataFileName.getCom());
                        waveDataMsg.setFrontType(dataFileName.getFrontType());
                        waveDataMsg.setFrontAccess(dataFileName.getFrontAccess());
                        waveDataMsg.setTemperature(dataFileName.getTemperature());
                        waveDataMsg.setFileType(dataFileName.getFileType());
                        waveDataMsg.setOriginalFile(fileName);
                        kafkaProducer.sendWaveData(waveDataMsg);
                    } else if (dataFileName.getFileType() == 6) {   // 文件类型为：6-故障，向”队列-故障“发送消息
                        // 此处设计有变更：1-不再传输故障文件数据、2-用指令0D来传输信息
                    } else if (dataFileName.getFileType() == 7) {   // 文件类型为：7-MVB，向”队列-MVB“发送消息
                        /*MvbDataMsg mvbDataMsg = new MvbDataMsg();
                        mvbDataMsg.setEquipmentNo(dataFileName.getEquipmentId());
                        mvbDataMsg.setFileType(dataFileName.getFileType());
                        mvbDataMsg.setOriginalFile(fileName);
                        kafkaProducer.sendMvbFileData(mvbDataMsg);*/
                    } else {
                        return;
                    }
                }
                return;
            }
        }
        /* 设备版本信息获取 (86) */
        if (fc.equals(AgreementCommandEnum.R_86.getValue())) {
            // 收到：前置版本信息
            if (agreement.getData().startsWith("01")) {
                FrontData frontData = frontHandler.getFrontInfoApplyData(agreement.getAddr(), agreement.getData());
                redisService.setCacheObject(agreement.getAddr() + "-FrontData", frontData, dataCacheTime, TimeUnit.MINUTES);
                return;
            }
            // 收到：DSP版本信息
            if (agreement.getData().startsWith("02")) {
                DspData dspData = frontHandler.getDspInfoApplyData(agreement.getAddr(), agreement.getData());
                redisService.setCacheObject(agreement.getAddr() + "-DspData", dspData, dataCacheTime, TimeUnit.MINUTES);
                return;
            }
            // 收到：主机（软硬件）版本信息
            if (agreement.getData().startsWith("03")) {
                VersionData versionData = frontHandler.getVersionApplyData(agreement.getAddr(), agreement.getData());
                redisService.setCacheObject(agreement.getAddr() + "-VersionData", versionData, dataCacheTime, TimeUnit.MINUTES);
                kafkaProducer.sendEquipmentVersion(versionData);
                return;
            }

            return;
        }
        /* 自检模式#回复 (87) */
        if (fc.equals(AgreementCommandEnum.R_87.getValue())) {
            // 收到：自检启动/关闭#回复
            if (agreement.getData().startsWith("01")) {
                CheckApply checkApply = checkHandler.getCheckCommandApply(agreement.getAddr(), agreement.getData());
                return;
            }
            return;
        }
        /* 自检模式#上报数据 (07) */
        if (fc.equals(AgreementCommandEnum.S_07.getValue())) {
            // 收到：自检模式-数据
            if (agreement.getData().startsWith("02")) {
                CheckData checkData = checkHandler.getCheckReportData(agreement.getAddr(), agreement.getData());
                // 回复响应
                checkHandler.replyCheckReportData(agreement, channel);

                // 缓存上报的数据
                String cacheKey = agreement.getAddr() + "-CheckData";
                redisService.setCacheList(cacheKey, checkData, RedisDataDirectionEnum.LEFT.getValue());
                redisService.expire(cacheKey, 1, TimeUnit.MINUTES);

                List<Object> lstData = redisService.getCacheList(cacheKey);

                // 调用远程方法，保存：自检模式-数据
                // businessTrainService.saveCheckData(checkData);
                return;
            }
            return;
        }
        /* MVB数据 （08） */
        if (fc.equals(AgreementCommandEnum.S_08.getValue())) {
            // 接收：MVB全量数据（车辆上电接收一次）
            if (agreement.getData().startsWith("01")) {
                MvbData mvbData = mvbHandler.getReportData(agreement.getAddr(), agreement.getData());
                mvbData.setOriginalData(agreement.getData());
                // log.warn("收到MVB-原始数据: {}", mvbData.getOriginalData());

                // 回复响应
                mvbHandler.replyGetReportData(agreement, channel);

                // 向Kafka发送消息
                kafkaProducer.sendMvbData(mvbData);
                return;
            }
            return;
        }
        /* MVB信息#回复 （88） */
        if (fc.equals(AgreementCommandEnum.R_88.getValue())) {
            // 收到：发送MVB测试数据的回复信息
            if (agreement.getData().startsWith("05")) {
                // log.warn("收到：发送MVB测试数据的回复信息 - {}", agreement.getData());
            }
            return;
        }
        /* 升级 (89) */
        if (fc.equals(AgreementCommandEnum.R_89.getValue())) {
            // 收到消息: 下发升级内容#回复
            if (agreement.getData().startsWith("01")) {
                UpgradeFileApply upgradeFileApply = upgradeHandler.getUpgradeFileApply(agreement.getAddr(), agreement.getData());
                String contentCacehKey = agreement.getAddr() + "#" + dataCacheTool.getUpgradeFileType(upgradeFileApply.getType());
                String rateCacehKey = agreement.getAddr() + "#" + dataCacheTool.getUpgradeFileType(upgradeFileApply.getType());

                // 发送升级内容-不成功，重新发送    TODO: 服务器连续3次收到异常应答，则停止本次升级
                if (upgradeFileApply.getResult() == 1) {
                    log.warn("发送升级内容-不成功，重新发送！");
                    String content = upgradeHandler.getUpgradeFileContentFromCache(contentCacehKey, rateCacehKey);
                    AgreementStructure upgradeAgreement = upgradeHandler.getUpgradeAgreement(agreement.getAddr(), content);
                    upgradeHandler.sendUpgradeFileData(upgradeAgreement, channel);
                    return;
                }

                // 发送升级文件内容未结束，继续发送
                if (upgradeFileApply.getUpgradeEndFlag() == 0) {
                    int upgradeFileRate = dataCacheTool.getUpgradeFileRateCache(rateCacehKey);
                    upgradeFileRate++;
                    dataCacheTool.putUpgradeFileRateCache(rateCacehKey, upgradeFileRate);

                    UpgradeProgress upgradeProgress = redisService.getCacheObject(
                            agreement.getAddr() + "-UpgradeFile-" + upgradeFileApply.getType()
                    );
                    if (upgradeProgress != null) {
                        upgradeProgress.setNumber(upgradeFileRate);
                        redisService.setCacheObject(agreement.getAddr() + "-UpgradeFile-" + upgradeFileApply.getType(),
                                upgradeProgress, dataCacheTime, TimeUnit.MINUTES);
                    }
                    String content = upgradeHandler.getUpgradeFileContentFromCache(contentCacehKey, rateCacehKey);
                    AgreementStructure upgradeAgreement = upgradeHandler.getUpgradeAgreement(agreement.getAddr(), content);
                    upgradeHandler.sendUpgradeFileData(upgradeAgreement, channel);
                    return;
                } else if (upgradeFileApply.getUpgradeEndFlag() == 1) {
                    // 升级结束，清空：升级文件缓存内容
                    log.warn("下发升级指令：升级结束");
                    dataCacheTool.removeUpgradeFileContentCache(contentCacehKey);
                    dataCacheTool.removeUpgradeFileRateCache(rateCacehKey);

                    UpgradeProgress upgradeProgress = redisService.getCacheObject(
                            agreement.getAddr() + "-UpgradeFile-" + upgradeFileApply.getType()
                    );
                    if (upgradeProgress != null) {
                        upgradeProgress.setNumber(upgradeProgress.getCount());
                        redisService.setCacheObject(agreement.getAddr() + "-UpgradeFile-" + upgradeFileApply.getType(),
                                upgradeProgress, dataCacheTime, TimeUnit.MINUTES);
                    }
                    return;
                }
            }
            // 收到消息: 返回上一个程序版本#回复
            if (agreement.getData().startsWith("02")) {
                /*UpgradeVersionApply version = upgradeHandler.getPreviousVersionApply(agreement.getAddr(), agreement.getData());
                redisService.setCacheObject(agreement.getAddr() + "-PreviousVersion", version, dataCacheTime, TimeUnit.MINUTES);*/
                return;
            }
            // 收到消息: 重启#回复
            if (agreement.getData().startsWith("03")) {
                UpgradeCommonApply upgradeCommonApply = upgradeHandler.getUpgradeCommonApply(agreement.getAddr(), agreement.getData());
                redisService.setCacheObject(agreement.getAddr() + "-Restart", upgradeCommonApply, dataCacheTime, TimeUnit.MINUTES);

                // 保存重启日志
                businessTrainService.saveRestartLogs(agreement.getAddr(), new Date(), upgradeCommonApply.getResult());
                return;
            }
            // 收到消息: 设备升级(1-前置、2-DSP)#回复
            if (agreement.getData().startsWith("04")) {
                UpgradeCommonApply upgradeCommonApply = upgradeHandler.getUpgradeCommonApply(agreement.getAddr(), agreement.getData());
                redisService.setCacheObject(agreement.getAddr() + "-EquipmentUpgrade", upgradeCommonApply, dataCacheTime, TimeUnit.MINUTES);
                return;
            }
        }
        /* 上报信息 (OD) */
        if (fc.equals(AgreementCommandEnum.S_0D.getValue())) {
            // 收到：设备异常
            if (agreement.getData().startsWith("01")) {
                EquipmentFault equipmentFault = faultHandler.getEquipmentFault(agreement.getAddr(), agreement.getData());
                EquipmentExceptionMsg equipmentExceptionMsg = new EquipmentExceptionMsg();
                equipmentExceptionMsg.setEquipmentNo(equipmentFault.getEquipmentId());
                equipmentExceptionMsg.setHostFaultStatus(equipmentFault.getHostFaultStatus());
                equipmentExceptionMsg.setDspFault(equipmentFault.getDspFault());
                equipmentExceptionMsg.setFront1Fault(equipmentFault.getFront1Fault());
                equipmentExceptionMsg.setFront1Type(equipmentFault.getFront1Type());
                equipmentExceptionMsg.setFront1Sensor(equipmentFault.getFront1Sensor());
                equipmentExceptionMsg.setFront2Fault(equipmentFault.getFront2Fault());
                equipmentExceptionMsg.setFront2Type(equipmentFault.getFront2Type());
                equipmentExceptionMsg.setFront2Sensor(equipmentFault.getFront2Sensor());
                equipmentExceptionMsg.setFront3Fault(equipmentFault.getFront3Fault());
                equipmentExceptionMsg.setFront3Type(equipmentFault.getFront3Type());
                equipmentExceptionMsg.setFront3Sensor(equipmentFault.getFront3Sensor());
                equipmentExceptionMsg.setFront4Fault(equipmentFault.getFront4Fault());
                equipmentExceptionMsg.setFront4Type(equipmentFault.getFront4Type());
                equipmentExceptionMsg.setFront4Sensor(equipmentFault.getFront4Sensor());

                // 回复
                faultHandler.reply01(agreement, channel);

                // 向Kafka发送消息
                kafkaProducer.sendEquipmentFault(equipmentExceptionMsg);
                return;
            }
            // 收到：检测故障
            if (agreement.getData().startsWith("02")) {
                // log.warn("收到 {} 检测故障：{}", agreement.getAddr(), agreement.getData());

                // 回复
                faultHandler.reply01(agreement, channel);
                return;
            }
            // 收到：MVB到站信息
            if (agreement.getData().startsWith("03")) {
                // log.warn("收到 {} MVB到站信息：{}", agreement.getAddr(), agreement.getData());
                MvbStation mvbStation = mvbHandler.getMvbStation(agreement.getAddr(), agreement.getData());
                mvbStation.setEquipmentNo(mvbStation.getEquipmentId());
                // log.warn("转换后的MVB到站信息: {}", mvbStation.toString());

                // 回复
                faultHandler.reply03(agreement, channel);

                // 向Kafka发送消息
                kafkaProducer.sendMvbStation(mvbStation);
                return;
            }
            // 收到：算法检测到的部件故障
            if (agreement.getData().startsWith("04")) {
                // log.info("收到 {} 算法检测到的部件故障：{}", agreement.getAddr(), agreement.getData());
                PartFault partFault = faultHandler.getPartFault(agreement.getAddr(), agreement.getData());
                PartFaultMsg partFaultMsg = new PartFaultMsg();
                partFaultMsg.setEquipmentNo(partFault.getEquipmentId());
                partFaultMsg.setFaultType(partFault.getFaultType());
                partFaultMsg.setTime(partFault.getTime());
                partFaultMsg.setCom(partFault.getCom());
                partFaultMsg.setFrontType(partFault.getFrontType());
                partFaultMsg.setFrontAccess(partFault.getFrontAccess());
                partFaultMsg.setFaultCount(partFault.getFaultCount());
                partFaultMsg.setDetails(partFault.getDetails());
                partFaultMsg.setOriginalDataFilename(partFault.getOriginalDataFilename());

                // 回复
                faultHandler.reply04(agreement, channel);

                // 向Kafka发送消息
                kafkaProducer.sendPartFault(partFaultMsg);
                return;
            }
            // 收到运行信息：ODOD  （2024-07-17增加）
            if (agreement.getData().toUpperCase().startsWith("0D")) {
                // log.info("收到 {} 上报的站点运行信息：{}", agreement.getAddr(), agreement);
                // log.info("数据区内容：{}", agreement.getData());
                TravelStationRegionData travelStationRegionData = faultHandler.getTravelStationRegionData(agreement.getAddr(), agreement.getData());

                // 回复
                AgreementStructure replyAgreement = faultHandler.reply0D(agreement, channel);
                // log.info("回复 {} 上报的站点运行信息：{}", replyAgreement.getAddr(), replyAgreement);

                // 向Kafka发送消息
                kafkaProducer.sendTravelStationRegion(travelStationRegionData);
                // log.info("向Kafka发送消息：{}", JSONObject.toJSONString(travelStationRegionData));
                return;
            }

            return;
        }

        log.error("错误的协议命令: {}，丢弃!", agreement.getFc());
        return;
    }


}